//package org.zfes.snowier.xyz.analysis.queue;
//
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.concurrent.LinkedBlockingQueue;
//import java.util.concurrent.ScheduledExecutorService;
//import java.util.concurrent.ThreadFactory;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicBoolean;
//import java.util.concurrent.locks.ReentrantLock;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.zfes.snowier.common.eventpub.EventBusBucket;
//import com.google.common.util.concurrent.ThreadFactoryBuilder;
//
//public class RedisAnalyQueueBucket  implements AnalyQueueBucket{
//
//	private static final String key_prefix="mo_li_";
//	
//	@Autowired
//	private RedisTemplate<String,Object> redisTemplate;
//	
//	private static ExecutorService postExecutorService;
//	
//	private static ExecutorService offerExecutorService;
//	
//	private volatile  AtomicBoolean runScheduled = new AtomicBoolean(false);
//	
//	private static AtomicBoolean sheduelPoolStarted = new AtomicBoolean(false);
//	
//	private static final ReentrantLock lock = new ReentrantLock();
//	
//	/**
//	 * 0.避免while-true检测，让队列有适当消息堆积能力
//	 * 1.固定时间频率
//	 * 2.队列超过长度
//	 */
//	private static ScheduledExecutorService scheduledPostExecutorService;
//	
//	@Override
//	public  void offer(Object msg) {
//		startPostSheduel();
//		getOfferPool().submit(()->redisTemplate.opsForList().leftPush(key_prefix,msg));
//	}
//	
//	
//	@Override
//	public void clear() {
//		redisTemplate.delete(key_prefix);
//	}
//	
//	
//	private  void startPostSheduel(){
//		lock.lock();
//		try {
//		
//			if(sheduelPoolStarted.get()) {
//				return;
//			}
//			if(postExecutorService==null) {
//				postExecutorService=Executors.newSingleThreadExecutor();
//			}
//			postExecutorService.submit(()->{
//				while(true) {
//					try {
//						 Long size=redisTemplate.opsForList().size(key_prefix);
//						 size=size==null?0:size;
//							if(size > SCHEDULED_QUEUE_LENGTH) {
//								runScheduled.set(false);
//								doPoll(defaultTopic);
//							}else {
//								if(!runScheduled.get()) {
//									runScheduled.set(true);
//								}
//							}
//						
//						
//					}catch (Exception e) {
//						e.printStackTrace();
//					}
//				}
//			});
//			
//			scheduledPostExecutorService.scheduleAtFixedRate(()->{
//				try {
//					if(runScheduled.get()) {
//						doPoll(defaultTopic,batchSize);
//					}
//				}catch (Exception e) {
//					e.printStackTrace();
//				}
//				
//			},1,100,TimeUnit.MILLISECONDS);
//			
//			sheduelPoolStarted.set(true);
//		}finally {
//			lock.unlock();
//		}
//	}
//	
//	private void doPoll(String topic,int batchSize) {
//		for(int i=0; i<batchSize; i++) {
//			Object data = redisTemplate.opsForList().rightPop(key_prefix, 0, TimeUnit.NANOSECONDS);
//			if(data!=null) {
//				return;
//			}else {
//				EventBusBucket.post(topic, new BasicMsg(data).setTopic(topic));
//			}
//			
//		}
//	}
//	
//	private void doPoll(String key) {
//		Object data = redisTemplate.opsForList().rightPop(key_prefix, 0, TimeUnit.NANOSECONDS);
//		if(data!=null) {
//			EventBusBucket.post(key, new BasicMsg(data).setTopic(key));
//		}
//	}
//	
//	private static ExecutorService  getOfferPool() {
//		if(offerExecutorService != null) {
//			return offerExecutorService;
//		}
//		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-queueBucket-OfferPool-pool-%d").build();
//		offerExecutorService = new ThreadPoolExecutor(0, maxOfferThreads,0L, 
//						TimeUnit.MILLISECONDS,
//						new LinkedBlockingQueue<Runnable>(maxOfferThreadQueueSize), 
//						namedThreadFactory, 
//						new ThreadPoolExecutor.AbortPolicy());
//		return offerExecutorService;				
//		
//	}
//
//	
//	
//}
